// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: gekim

#include "coin2/exchange/okex_common/feed_v5/processor.h"

#include <rapidjson/error/en.h>

#include <string_view>
#include <utility>

#include "coin2/base/string_util.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/okex_common/feed_v5/update.h"
#include "coin2/exchange/util/json_util.h"

using namespace std::literals;

namespace coin2::exchange::okex_common::feed_v5::impl {

using coin2::exchange::okex_common::feed_v5::OkexCommonFeedUpdateParser;

bool OkexCommonFeedProcessor::CheckChannelFilter(const std::string& ch) {
  // index is rest api, its format is different with websocket api
  if (FeedParsingProcessor::CheckChannelFilter(std::string(ch))) {
    return true;
  } else {
    return FeedParsingProcessor::CheckChannelFilter(std::string(ch) + ":{}");
  }
}

// we share the same book builder between depth5, depth400 snapshot, depth400
// diff
void OkexCommonFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  if (packet.payload == "pong") return;
  upd_mgr_.onBeginPacket(packet);

  simdjson::padded_string buffer(packet.payload);
  simdjson::ondemand::document document;
  auto error = parser_.iterate(buffer).get(document);
  if (error) {
    throw std::runtime_error(
        "[Okex Common Feed Processor] JSON parse error: "s +
        simdjson::error_message(error));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(packet.payload);
  }

  std::string_view event_string;
  auto event_err = document["event"].get_string().get(event_string);
  if (!event_err) {
    if (event_string == "subscribe"sv) {
      // TODO: check subscribe result.
    } else if (event_string == "login"sv) {
      std::string_view code_string = document["code"];
      if (code_string != "0"sv) {
        throw std::runtime_error(
            "[Okex Common Feed Processor] login error: "s +
            packet.payload.data());
      }
    }
    upd_mgr_.onEndPacket(packet);
    return;
  }

  bool is_channel_data = false;
  bool is_rest_data = false;
  simdjson::ondemand::object json_arg;
  std::string_view str_channel_type;
  std::string_view native_symbol;
  std::string_view action;
  std::string_view url;
  auto arg_err = document["arg"].get(json_arg);
  if (arg_err == simdjson::SUCCESS) {
    auto channel_err = json_arg["channel"].get(str_channel_type);
    auto symbol_err = json_arg["instId"].get(native_symbol);
    if (channel_err == simdjson::SUCCESS &&
        symbol_err == simdjson::SUCCESS) {
      is_channel_data = true;
    }

    document["action"].get(action);
  } else {
    auto url_err = document["url"].get_string().get(url);
    if (url_err == simdjson::SUCCESS) is_rest_data = true;
  }
  const std::string native_symbol_str(native_symbol.begin(), native_symbol.end());
  simdjson::ondemand::value json_data = document["data"];

  if (is_channel_data) {
    std::string channel(str_channel_type.data(), str_channel_type.size());
    if (!CheckChannelFilter(channel)) {
      return;
    }

    if (str_channel_type == "trades") {
      if (parser_config_.has_trade_sampling_frequency_bps()) {
        if (upd_mgr_.is_symbol_registered(native_symbol)) {
          auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
          for (simdjson::ondemand::object json_trade : json_data.get_array()) {
            Trade _trade{};
            OkexCommonFeedUpdateParser::ParseTrade(json_trade, &_trade);
            bool aggregated = slim_trade->TryAggregateSlimTrade(
                    packet.timestamp,
                    _trade.price,
                    _trade.fill_qty,
                    _trade.side,
                    _trade.timestamp,
                    parser_config_.trade_sampling_frequency_bps());
            int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
            if (!aggregated ||
                timespan > parser_config_.trade_sampling_time_length()) {
              auto* trade = upd_mgr_.GetTradeMutable();
              CHECK(slim_trade->PopulateTradeAndClear(trade));
              upd_mgr_.PublishTradeWithExchangeTimeCheck(native_symbol_str, false);
            }
          }
        }
      } else {
        std::string last_symbol;

        Trade* agg_trade = upd_mgr_.GetTradeMutable();
        agg_trade->Clear();

        Trade trade{};
        for (simdjson::ondemand::object json_trade : json_data.get_array()) {
          if (!upd_mgr_.is_symbol_registered(native_symbol)) {
            continue;
          }
          OkexCommonFeedUpdateParser::ParseTrade(json_trade, &trade);
          if (native_symbol == last_symbol &&
              TryAggregateTrades(agg_trade, trade)) {
            continue;
          }
          if (agg_trade->fill_qty > 0) {
            CHECK_GT(last_symbol.size(), 0);
            upd_mgr_.PublishTradeWithExchangeTimeCheck(last_symbol, true);
          }
          *agg_trade = trade;
          last_symbol = native_symbol;
        }
        if (agg_trade->fill_qty > 0) {
          upd_mgr_.PublishTradeWithExchangeTimeCheck(last_symbol, false);
        }
      }
    } else if (str_channel_type == "bbo-tbt") {
      for (simdjson::ondemand::object json_book : json_data.get_array()) {
        auto book = upd_mgr_.GetBookMutable(native_symbol, "bbo");
        if (book != nullptr) {
          if (publish_bbo_book_only_) {
            std::optional<PriceQty> bid_pre = book->Bid0();
            std::optional<PriceQty> ask_pre = book->Ask0();
            if (OkexCommonFeedUpdateParser::ParseBook(
                    book, json_book, BookBuilder::SNAPSHOT,
                    native_symbol, packet.timestamp)) {
              std::optional<PriceQty> bid_post = book->Bid0();
              std::optional<PriceQty> ask_post = book->Ask0();
              bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
              if (is_bbo_changed) {
                upd_mgr_.PublishBook();
              }
            }
          } else if (parser_config_.has_sampling_frequency_bps()) {
            if (OkexCommonFeedUpdateParser::ParseBook(
                    book, json_book, BookBuilder::SNAPSHOT,
                    native_symbol, packet.timestamp)) {
              bool sampled = book->SampleByFrequency(
                      book->Bid0(), book->Ask0(),
                      parser_config_.sampling_frequency_bps());
              if (parser_config_.has_trade_sampling_frequency_bps()) {
                auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
                int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
                if (sampled ||
                    timespan > parser_config_.trade_sampling_time_length()) {
                  auto* trade = upd_mgr_.GetTradeMutable();
                  if (slim_trade->PopulateTradeAndClear(trade)) {
                    upd_mgr_.PublishTradeWithExchangeTimeCheck(native_symbol_str, false);
                  }
                }
              }
              if (sampled) {
                upd_mgr_.PublishBook();
              }
            }
          } else {
            if (OkexCommonFeedUpdateParser::ParseBook(
                    book, json_book, BookBuilder::SNAPSHOT,
                    native_symbol, packet.timestamp)) {
              upd_mgr_.PublishBook();
            }
          }
        }
      }
    } else if (str_channel_type == "books5") {
      for (simdjson::ondemand::object json_trade : json_data.get_array()) {
        auto book = upd_mgr_.GetBookMutable(native_symbol, "depth5");
        if (book != nullptr) {
          if (publish_bbo_book_only_) {
            std::optional<PriceQty> bid_pre = book->Bid0();
            std::optional<PriceQty> ask_pre = book->Ask0();
            if (OkexCommonFeedUpdateParser::ParseBook(
                    book, json_trade, BookBuilder::LIGHT_BBO,
                    native_symbol, packet.timestamp)) {
              std::optional<PriceQty> bid_post = book->Bid0();
              std::optional<PriceQty> ask_post = book->Ask0();
              bool is_bbo_changed =
                  (bid_pre != bid_post) || (ask_pre != ask_post);
              if (is_bbo_changed) {
                upd_mgr_.PublishBook();
              }
            }
          } else {
            if (OkexCommonFeedUpdateParser::ParseBook(
                    book, json_trade, BookBuilder::DELTA,
                    native_symbol, packet.timestamp)) {
              upd_mgr_.PublishBook();
            }
          }
        }
      }
    } else if (str_channel_type == "books" ||
               str_channel_type == "books-l2-tbt") {
      if (action == "snapshot") {
        for (simdjson::ondemand::object json_books : json_data.get_array()) {
          auto book = upd_mgr_.GetBookMutable(native_symbol, "depth");
          if (book != nullptr) {
            if (parser_config_.has_sampling_frequency_bps()) {
              if (OkexCommonFeedUpdateParser::ParseBook(
                      book, json_books, BookBuilder::LIGHT_BBO,
                      native_symbol, packet.timestamp)) {
                bool sampled = book->SampleByFrequency(
                        book->Bid0(), book->Ask0(),
                        parser_config_.sampling_frequency_bps());
                if (parser_config_.has_trade_sampling_frequency_bps()) {
                  auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
                  int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
                  if (sampled ||
                      timespan > parser_config_.trade_sampling_time_length()) {
                    auto* trade = upd_mgr_.GetTradeMutable();
                    if (slim_trade->PopulateTradeAndClear(trade)) {
                      upd_mgr_.PublishTradeWithExchangeTimeCheck(native_symbol_str, false);
                    }
                  }
                }
                if (sampled) {
                  upd_mgr_.PublishBook();
                }
              }
            } else if (publish_bbo_book_only_) {
              std::optional<PriceQty> bid_pre = book->Bid0();
              std::optional<PriceQty> ask_pre = book->Ask0();
              if (OkexCommonFeedUpdateParser::ParseBook(
                      book, json_books, BookBuilder::LIGHT_BBO,
                      native_symbol, packet.timestamp)) {
                std::optional<PriceQty> bid_post = book->Bid0();
                std::optional<PriceQty> ask_post = book->Ask0();
                bool is_bbo_changed =
                    (bid_pre != bid_post) || (ask_pre != ask_post);
                if (is_bbo_changed) {
                  upd_mgr_.PublishBook();
                }
              }
            } else {
              if (OkexCommonFeedUpdateParser::ParseBook(
                      book, json_books, BookBuilder::SNAPSHOT,
                      native_symbol, packet.timestamp)) {
                upd_mgr_.PublishBook();
              }
            }
          }
        }
      } else if (action == "update") {
        for (simdjson::ondemand::object json_books : json_data.get_array()) {
          auto book = upd_mgr_.GetBookMutable(native_symbol, "depth");
          if (book != nullptr) {
            if (parser_config_.has_sampling_frequency_bps()) {
              if (OkexCommonFeedUpdateParser::ParseBook(
                      book, json_books, BookBuilder::LIGHT_BBO,
                      native_symbol, packet.timestamp)) {
                bool sampled = book->SampleByFrequency(
                        book->Bid0(), book->Ask0(),
                        parser_config_.sampling_frequency_bps());
                if (parser_config_.has_trade_sampling_frequency_bps()) {
                  auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
                  int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
                  if (sampled ||
                      timespan > parser_config_.trade_sampling_time_length()) {
                    auto* trade = upd_mgr_.GetTradeMutable();
                    if (slim_trade->PopulateTradeAndClear(trade)) {
                      upd_mgr_.PublishTradeWithExchangeTimeCheck(native_symbol_str, false);
                    }
                  }
                }
                if (sampled) {
                  upd_mgr_.PublishBook();
                }
              }
            } else if (publish_bbo_book_only_) {
              std::optional<PriceQty> bid_pre = book->Bid0();
              std::optional<PriceQty> ask_pre = book->Ask0();
              if (OkexCommonFeedUpdateParser::ParseBook(
                      book, json_books, BookBuilder::LIGHT_BBO,
                      native_symbol, packet.timestamp)) {
                std::optional<PriceQty> bid_post = book->Bid0();
                std::optional<PriceQty> ask_post = book->Ask0();
                bool is_bbo_changed =
                    (bid_pre != bid_post) || (ask_pre != ask_post);
                if (is_bbo_changed) {
                  upd_mgr_.PublishBook();
                }
              }
            } else {
              if (OkexCommonFeedUpdateParser::ParseBook(
                      book, json_books, BookBuilder::DELTA,
                      native_symbol, packet.timestamp)) {
                upd_mgr_.PublishBook();
              }
            }
          }
        }
      } else {
        Error("[Okex Common Processor] not implmeneted for ",
              packet.payload);
      }
    } else if (str_channel_type == "open-interest") {
      for (simdjson::ondemand::object json_open_interest : json_data.get_array()) {
        std::string_view native_symbol =
            json_open_interest["instId"].get_string();
        if (upd_mgr_.is_symbol_registered(native_symbol)) {
          auto* open_interest = upd_mgr_.GetOpenInterestMutable();
          if (OkexCommonFeedUpdateParser::ParseOpenInterest(json_open_interest,
                                                            open_interest)) {
            upd_mgr_.PublishOpenInterest(native_symbol);
          }
        }
      }
    } else if (str_channel_type == "mark-price") {
      for (simdjson::ondemand::object json_mark_price : json_data.get_array()) {
        std::string_view native_symbol = json_mark_price["instId"].get_string();
        if (upd_mgr_.is_symbol_registered(native_symbol)) {
          auto* mark_price = upd_mgr_.GetMarkPriceMutable();
          OkexCommonFeedUpdateParser::ParseMarkPrice(json_mark_price,
                                                     mark_price);
          upd_mgr_.PublishMarkPrice(native_symbol);
        }
      }
    } else if (str_channel_type == "funding-rate") {
      for (simdjson::ondemand::object json_funding_rate : json_data.get_array()) {
        std::string_view native_symbol =
            json_funding_rate["instId"].get_string();
        if (upd_mgr_.is_symbol_registered(native_symbol)) {
          auto* funding_rate = upd_mgr_.GetFundingRateMutable();
          OkexCommonFeedUpdateParser::ParseFundingRate(
              json_funding_rate, funding_rate, packet.timestamp);
          upd_mgr_.PublishFundingRate(native_symbol);
        }
      }
    } else if (str_channel_type == "index-tickers") {
      for (simdjson::ondemand::object json_index_tickers : json_data.get_array()) {
        std::string underlying_spot;
        auto* index = upd_mgr_.GetIndexMutable();
        OkexCommonFeedUpdateParser::ParseIndex(json_index_tickers, index,
                                               &underlying_spot);
        for (const auto& product : upd_mgr_.products()) {
          if (product->native_symbol().find(underlying_spot) !=
              std::string::npos) {
            upd_mgr_.PublishIndexWithExchangeTimeCheck(
                product->feed_native_symbol());
          }
        }
      }
    }
  } else if (is_rest_data) {
    // rest kline only support 1 worker, worker string is behind group
    if (!long_topic_.has_value()) {
      if (packet.topic_name == upd_mgr_.mea().SnakeString()) {
        long_topic_ = false;
      } else {
        long_topic_ = true;
      }
    }
    // only long topic contains worker setting
    if (long_topic_.value()) {
      int pos = packet.topic_name.find_last_of('_');
      int worker = std::stoi(packet.topic_name.substr(pos + 2));
      if (picked_worker_) {
        if (worker != picked_worker_) return;
      } else {
        picked_worker_ = worker;
      }
    }

    auto [native_symbol, bar, page] =
        OkexCommonFeedUpdateParser::ExplodeUrl(url);
    fastfeed::proto::KlineInterval interval =
        OkexCommonFeedUpdateParser::ParseKlineInterval(bar);

    if (!CheckChannelFilter(bar)) {
      return;
    }

    bool is_freezed_data = false;
    auto okex_channel = OkexCommonFeedUpdateParser::ParseOkxChannelPath(url);
    if (okex_channel == "history-candles") {
      is_freezed_data = true;
    }
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines != nullptr) {
      // rest api, kline is descending
      // query is split up several requests
      if (page == 0) {
        // first long query
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }

      simdjson::ondemand::array items = json_data["data"].get_array();
      const auto& quote = upd_mgr_.GetQuoteCurrency(native_symbol);
      bool is_adjust_turnover = true;
      if ((MarketType::Spot == upd_mgr_.mea().market) || (quote == "USD")) {
        is_adjust_turnover = false;
      }
      
      int64_t last_open_timestamp = 0;
      auto last_kline = klines->Get();
      if (last_kline) {
        last_open_timestamp = (*last_kline)->open_timestamp;
      }

      std::vector<Kline> kline_vec;
      for (simdjson::ondemand::value kline_data : items) {
        kline_vec.emplace_back(Kline());
        if (!OkexCommonFeedUpdateParser::Parse1Kline(
            kline_data, packet.timestamp, interval, is_adjust_turnover,
            last_open_timestamp, &kline_vec.back())) {
          kline_vec.pop_back();
          break;
        }
      }

      for (auto i = kline_vec.rbegin(); i < kline_vec.rend(); i++) {
        auto result = klines->AddKline(i->open_timestamp);
        CHECK(result);
        Kline* kline;
        bool freezed;
        std::tie(kline, freezed) = result.value();
        *kline = *i;
      }

      if (is_freezed_data) {
        klines->Freeze();
      }

      if (!kline_vec.empty()) {
        upd_mgr_.PublishKline(native_symbol);
      }
    }
  } else {
    LOG(ERROR) << "[Okex Common Feed Processor] extract data error "
                << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::okex_common::feed_v5::impl
